package com.example.spark.core

import com.example.util.SparkUtil
import org.apache.commons.lang3.StringUtils
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.util.LongAccumulator

/**
 * @title: RddOperate
 * @projectName bigdata
 * @description: RddOperate
 * @author leali
 * @date 2022/5/15 9:27
 */
object RddOperate {

  def operateAggregate(rdd: RDD[Int]): Unit = {
    println(s"rdd.sum() = ${rdd.sum()}")
    println(s"rdd.reduce(_ + _) = ${rdd.reduce((_: Int) + (_: Int))}")
    println(s"rdd.fold(0)(_ + _) = ${rdd.fold(0)((_: Int) + (_: Int))}")
    println(s"rdd.aggregate(0)(_ + _, _ + _) = ${rdd.aggregate(0)((_: Int) + (_: Int), (_: Int) + (_: Int))}")
  }

  /**
   * reduceByKey 都是分组+聚合，但是有预聚合
   * 因此，在对大数据进行复杂计算时，reduceByKey优于groupByKey。
   * 如果仅仅是group处理，那么以下函数应该优先于 groupByKey
   * >> combineByKey 组合数据，但是组合之后的数据类型与输入时值的类型不一样
   * >> foldByKey合并每一个 key 的所有值，在级联函数和“零值”中使用
   *
   * @param rdd rd
   */
  def operateAggregateByKey(rdd: RDD[(String, Int)]): Unit = {
    SparkUtil.printLog("groupByKey")
    rdd.groupByKey.mapValues((_: Iterable[Int]).sum).foreach(println)

    SparkUtil.printLog("reduceByKey")
    rdd.reduceByKey((_: Int) + (_: Int)).foreach(println)

    SparkUtil.printLog("foldByKey")
    rdd.foldByKey(0)((_: Int) + (_: Int)).foreach(println)

    SparkUtil.printLog("aggregateByKey")
    rdd.aggregateByKey(0)((_: Int) + (_: Int), (_: Int) + (_: Int)).foreach(println)
  }

  def operateJoin(rdd1: RDD[(Int, String)], rdd2: RDD[(Int, String)]): Unit = {
    SparkUtil.printLog("  rdd1.join(rdd2)")
    rdd1.join(rdd2).foreach(println)

    SparkUtil.printLog("  rdd1.leftOuterJoin(rdd2)")
    rdd1.leftOuterJoin(rdd2).foreach(println)

    SparkUtil.printLog("  rdd1.rightOuterJoin(rdd2)")
    rdd1.rightOuterJoin(rdd2).foreach(println)
  }

  def operateSort(rdd: RDD[(String, Int)]): Unit = {
    SparkUtil.printLog("sortBy")
    rdd.sortBy((_: (String, Int))._2, ascending = false).take(3).foreach(println)

    SparkUtil.printLog("sortByKey")
    rdd.map((_: (String, Int)).swap).sortByKey(ascending = false).take(3).foreach(println)

    SparkUtil.printLog("top")
    rdd.top(3)(Ordering.by((_: (String, Int))._2)).foreach(println)
  }

  /**
   * 注意:resultRDD在后续会被频繁使用到,且该RDD的计算过程比较复杂,所以为了提高后续访问该RDD的效率,应该将该RDD放到缓存中
   *
   * @param rdd (String,Int)
   */
  def operateCache(rdd: RDD[(String, Int)]): Unit = {
    //rdd.cache()//底层persist()
    //rdd.persist()//底层persist(StorageLevel.MEMORY_ONLY)
    rdd.persist(StorageLevel.MEMORY_AND_DISK)
    // watch run time
    operateSort(rdd)
  }

  /**
   * 上面的缓存持久化并不能保证RDD数据的绝对安全,所以应使用Checkpoint把数据发在HDFS上
   * 开发中-对于计算复杂且后续会被频繁使用的RDD先进行缓存/持久化,再进行Checkpoint
   * 1.存储位置
   * 缓存/持久化数据存默认存在内存, 一般设置为内存+磁盘(普通磁盘)
   * Checkpoint检查点:一般存储在HDFS
   * 2.功能
   * 缓存/持久化:保证数据后续使用的效率高
   * Checkpoint检查点:保证数据安全/也能一定程度上提高效率
   * 3.对于依赖关系:
   * 缓存/持久化:保留了RDD间的依赖关系
   * Checkpoint检查点:不保留RDD间的依赖关系
   *
   * @param sparkContext SparkContext
   * @param rdd          (String,Int)
   */
  def operateCheckpoint(sparkContext: SparkContext, rdd: RDD[(String, Int)]): Unit = {
    //rdd.cache()//底层persist()
    //rdd.persist()//底层persist(StorageLevel.MEMORY_ONLY)
    sparkContext.setCheckpointDir("/spark/checkpoint")
    rdd.persist(StorageLevel.MEMORY_AND_DISK)
    // watch run time
    rdd.checkpoint()
    //how to ?
    operateSort(rdd)
  }

  def operateSharedVariable(sparkContext: SparkContext): Unit = {
    val count: LongAccumulator = sparkContext.longAccumulator("count")
    //定义一个特殊字符集合
    val ruleList: List[String] = List(",", ".", "!", "#", "$", "%")
    //将集合作为广播变量广播到各个节点
    val broadcastRuleList: Broadcast[List[String]] = sparkContext.broadcast(ruleList)
    val value: RDD[(String, Int)] = sparkContext.textFile("src/data/input/rule_word.txt")
      .filter(StringUtils.isNoneBlank(_: String))
      .flatMap((_: String).split("\\s+"))
      .filter((ch: String) => {
        //获取广播数据
        if (broadcastRuleList.value.contains(ch)) {
          count.add(1)
          false
        } else {
          true
        }
      })
      .map((_: String, 1))
      .reduceByKey((_: Int) + (_: Int))

    // print
    value.foreach(println)
    println(s"count.value = ${count.value}")
  }

  def operateExternalDataSource(sparkContext: SparkContext): Unit = {
    val value: RDD[(String, Int)] = sparkContext.textFile("src/data/input/word.txt")
      .filter(StringUtils.isNoneBlank(_: String))
      .flatMap((_: String).split(" "))
      .map((_: String, 1))
      .reduceByKey((_: Int) + (_: Int))
    value.repartition(1).saveAsTextFile("data/output/result1")
    value.repartition(1).saveAsObjectFile("data/output/result2")
    value.repartition(1).saveAsSequenceFile("data/output/result3")
  }
}
